package org.NooLab.glue.subscription;

import java.util.*;

import org.NooLab.compare.utilities.math.DistanceControlProperties;

import org.NooLab.utilities.logging.PrintLog;
import org.NooLab.utilities.net.GUID;
import org.NooLab.utilities.strings.StringsUtil;

import org.NooLab.glue.MessageBoard;
import org.NooLab.glue.components.*;
import org.NooLab.glue.msgframe.*;
import org.NooLab.glue.subscription.context.CompareResultHandle;
import org.NooLab.glue.subscription.context.CompareResultHandleIntf;
import org.NooLab.glue.subscription.context.ComparisonIntf;
import org.NooLab.glue.subscription.context.Context;
import org.NooLab.glue.subscription.context.ContextInfra;
import org.NooLab.glue.storage.MsgBoardDataStorage;
import org.NooLab.glue.transaction.Transaction;

 


public class MessageMatcherProcess implements Runnable{

	// =================================

	// object references ..............
	
	MessageBoard messageBoard ;
	MetaRegistration metaRegistration;
	MessageBoardFactory mbf;
	
	
	MessageMatcherIntf msgMatcher; 
 	
	Vector<XRegistration> registeredInstances;
	ClientMatcherIntf clientMatcher;
	
	FuturesIntf futures;
	CompareResultHandleIntf compareResultHandle ;
	
	protected XMessage xmsg;
	protected CtxtXMessage cxmsg;
	// main variables / properties ....

	Vector<MMMessage> contextDescriptions = new Vector<MMMessage> ();
	
	public boolean __DEBUG_SHP_running = false;
	 
	// constants ......................
	
	

	// volatile variables .............

	Thread msgMThrd;
	boolean isrunning = false;
	boolean isworking = false;
	
	int processCounter=0;
	int maxProcessCount = 10;
	
	// helper objects .................
	 
	PrintLog out;
	StringsUtil strgutil = new StringsUtil (); 
	
	// . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
	
	public MessageMatcherProcess( MessageBoard msgBoard, MessageMatcherIntf msgmatcher , PrintLog outprn){
		
		messageBoard = msgBoard;
		msgMatcher = msgmatcher;

		mbf = messageBoard.getMsgBoardShares().getMsgBoardFactory() ;
		
		clientMatcher = messageBoard.getClientMatcher() ;
		
		metaRegistration = messageBoard.getMetaRegistration() ;
		registeredInstances = metaRegistration.getRegisteredInstances() ; 
		
		// futures are retrieved by "SubscriptionHandlerInterpreter{}"
		futures = messageBoard.getFutures();
		
		compareResultHandle = new CompareResultHandle();
		
		out = outprn;
		xmsg = new XMessage(out);
		cxmsg = new CtxtXMessage(out);
		
		msgMThrd = new Thread(this, "MessageMatcherProcess") ;
		msgMThrd.start();
	}

	@Override
	public void run() {
		
		isrunning = true;
		
		try{
		
			while (isrunning){
				
				if (__DEBUG_SHP_running){
					out.delay(2000) ;
					continue;
				}
				if ((contextDescriptions.size()>0) && (processCounter<= maxProcessCount)){
					// each new arriving message opens a new process, to make subscription handling
					// independent from this loop
					(new SubscriptionHandlingProcess( this, contextDescriptions.get(0))).start() ;
					contextDescriptions.remove(0) ;
				} // any msg to work on ?
				
				out.delay(10) ;
			} // while ->
			
			
		}catch(Exception e){
			
		}
		
	} // run

	public void stopMMProcess(){
		
		isrunning = false;
		
	}

	/**
	 * 
	 * only SOURCES should call this, the xml message has to contain a "context" - section !
	 * (actually the message consists ONLY from the context section...)
	 * 
	 * @param rawXmlMsg
	 * @return
	 */
	public String setMessageContent( String rawXmlMsg ){ // this xml is WITHOUT encoded transaction data object !!
	
		MMMessage mmm;
		String guidStr="";
		
		
		mmm = new MMMessage();
		
		// only if the issuer discloses its iid and tid in the message...
		mmm.iid = -1;
		mmm.tid = -1;
		
		// is there a context section in the message
		if ((rawXmlMsg.contains("<context")==false) || (rawXmlMsg.contains("<pragma")==false)){			 
			return "";
		}
												out.print(2,"");
		// using a dedicated process to prepare the xml, and immediately 
		// returning to dispatcher of MessageBoard...
        AssimilateRawXml arx = new AssimilateRawXml(rawXmlMsg, mmm) ;
        
        while (arx.getMMMGuid().length()==0){
        	out.delay(1);
        }
        guidStr = arx.getMMMGuid(); 
        
		return guidStr;
	}

	/**
	 * 
	 * use this for checking whether the given instanceID and the provided transaction match...
	 * 
	 * @param instanceID
	 * @param transaction
	 * @param targetedRole
	 * @param contentType
	 * @return
	 */
	public long performMatchingForInstance( long instanceID, Transaction transaction, 
											String targetedRole , String contentType){
		long transactionID=-1;
		long  iid=0,  xtid, xiid,tid ;
		boolean rmb;
		int ix;
		CClients candidateClients = new CClients( mbf ); 
		long[] iids;
		Transaction transact;
		XRegistration xxreg ;
		Vector<Future> futs;
		Future future;
		
	 
		
		ContextHeaderDescription cd = null, contextDescription = null;
		ActiveContextHeaderDescriptions contextDescriptions;
		
		// we first check the doc type, group settings, since this is less expensive
		
		xtid = transaction.getTransactionID() ;
		xiid = transaction.getInstanceID() ;
		// the originator of the transaction might not be registered...
		xxreg = XRegistration.getItemByID( xiid, registeredInstances);
		
		// creating a copy of the object
		transact = new Transaction( transaction ) ;
		
		// removing the role identifier, in order to check ALL clients, not just the counterparts...
		// later, when returning a port, we have to restrict to the counterpart !!
		transact.setRoleOfInitializer("") ;
		transact.setDataObject(null); // not the data are deleted, just the copy of the pointer is being reset
		
		
		// retrieving a matching list, essentially, just the doc type and the group settings apply so far
		// remove the instance that issued the transaction from the list of candidates
	
		iids = clientMatcher.findMatchingParticipants( transact, 0 ) ;  
			   Arrays.sort(iids);
			   
			   ix = Arrays.binarySearch(iids, xiid);
			   if (ix>=0){
				   iids[ix] = -1;
			   }
		
		// is our provided instance in the list of candidates ?
	    ix = Arrays.binarySearch(iids, instanceID);
	         if (ix<0){
	        	 return -3;
	         }
			 
	    iids = new long[1] ;
	    iids[0] = instanceID;
	    
	    
		contextDescriptions = messageBoard.getActiveContextHeaderDescriptions();
		 
		// selecting and removing the partner client, if the transaction was a task
		// int ttype = transaction.getType();
		// is data type
		targetedRole = targetedRole.toLowerCase().trim();
		
		// select the context info issued by this transaction   org.NooLab.glue.subscription.Futures@1a2467a
		for (int c=0; c < contextDescriptions.size();c++){
			cd = contextDescriptions.getItem(c);
			
			if (cd.transactionID == xtid){
				contextDescription = cd;
				break;
			}
		}
		
		 
		 
		if (contextDescription==null){
			return -3;
		}
		
		// all candidates as identified by iid, (1) is there a subscription issued by it? (2) if yes, does it match?
		for (int i=0;i<iids.length;i++){

			
			iid = iids[i] ;
			if ((iid==xiid) || (iid<0)){
				continue;
			}

 
			
			futs = futures.getFuturesById( iid ); // regardless the type (_PLAIN, _CONTEXT)
			
			for (int s=0;s<futs.size(); s++){
				future = futs.get(s);	
												out.print(3,"going to perform next comparison between contextDescription, future...");
				rmb = matchingContextInfoToSubscription( contextDescription, future );
				
				if (rmb){
					tid = contextDescription.transactionID ;
					candidateClients.add( tid, future) ;
				}
				
			} // s->
			 
			// xRegCandidates
		} // i->
		
		 
		// there should be only 1
		tid = -1;
		if (candidateClients.instanceIDs.size()>0){
			tid = candidateClients.instanceIDs.get(0) ;
		}
		
		candidateClients.clear();
		
		return tid ;

	}
	
 
	
	// AS SUBSCRIPTION (NOT: a "task") to subscribers !
    // if not (), but instead with parameters, then one of the subscribers is de-selected and 
	// instead returned to here by means of its iid... 
	/**
	 * ATTENTION: this is often called from Receptor context, i.e. from "ReceptorMessageInterpreter{}"
	 * checking whether there is any subscription matching the context info contain in  the transaction tid 
	 * 
	 */
	public long performMatchingRaid( long rqtid, String role , String contentType){
		long instanceID =0;
		
		// we create a VERY basic transaction instance ...
		
		return instanceID ;
	}
	
	/** 
	 * 
	 * TODO: we some kind of cache, for already tested context/futures pairs, and even content/futures pairs
	 *
	 * called e.g. by incoming transaction taskpackage issued by a SOURCE client, 
	 * i.e. from "SourceMessageInterpreter{}"
	 * 
	 * looking for all clients matching role and doc type,..
	 * then, within this set, select 
	 */
	public long performMatchingRaid( Transaction transaction, String targetedRole , String contentType){
							                          //       "RECEPTOR"   "doc"
		long instanceID=-1, iid=0,  xtid, xiid ;
		boolean  rmb;
		int ix;
		CClients candidateClients = new CClients( mbf ); 
		long[] iids;
		Transaction transact;
		XRegistration xxreg ;
		Vector<Future> futs;
		Future future;
		 
		
		ContextHeaderDescription cd = null, contextDescription = null;
		ActiveContextHeaderDescriptions contextDescriptions;
		
		// we first check the doc type, group settings, since this is less expensive
		
		xtid = transaction.getTransactionID() ;
		xiid = transaction.getInstanceID() ;
		
		xxreg = XRegistration.getItemByID( xiid, registeredInstances);
		
		// creating a copy of the object
		transact = new Transaction( transaction ) ;
		
		// removing the role identifier, in order to check ALL clients, not just the counterparts...
		// later, when returning a port, we have to restrict to the counterpart !!
		transact.setRoleOfInitializer("") ;
		transact.setDataObject(null); // not the data are deleted, just the copy of the pointer is being reset
		
		
		// retrieving a matching list, essentially, just the doc type and the group settings apply so far
		// remove the instance that issued the transaction from the list of candidates
	
		iids = clientMatcher.findMatchingParticipants( transact, 0 ) ;  
			   Arrays.sort(iids);
			   ix = Arrays.binarySearch(iids, xiid);
			   if (ix>=0){
				   iids[ix] = -1;
			   }
		
		contextDescriptions = messageBoard.getActiveContextHeaderDescriptions();
		 
		// selecting and removing the partner client, if the transaction was a task
		// int ttype = transaction.getType();
		// is data type
		targetedRole = targetedRole.toLowerCase().trim();
		
		// select the context info issued by this transaction   org.NooLab.glue.subscription.Futures@1a2467a
		for (int c=0; c < contextDescriptions.size();c++){
			cd = contextDescriptions.getItem(c);
			
			if (cd.transactionID == xtid){
				contextDescription = cd;
				break;
			}
		}
		
		if (contextDescription==null){
			return -3;
		}
		
		// all candidates as identified by iid, (1) is there a subscription issued by it? (2) if yes, does it match?
		for (int i=0;i<iids.length;i++){

			
			iid = iids[i] ;
			if ((iid==xiid) || (iid<0)){
				continue;
			}
			
			// boolean hb = true; // indicator, trying to refute...  
			
			futs = futures.getFuturesById( iid, -1); // -1 -> regardless the type (_PLAIN, _CONTEXT)
			
			for (int s=0;s<futs.size(); s++){
				future = futs.get(s);	
												out.print(3,"going to perform next comparison between contextDescription, future...");
				rmb = matchingContextInfoToSubscription( contextDescription, future );
				
				if (rmb){
					candidateClients.add(iid, future) ;
				}
				
			} // s->
			 
			// xRegCandidates
		} // i->
		
		// this identifies a RECEPTOR for the message ONLY if the transactions requests
		// that subscription send-outs are based on matching subscription
		iid = findReceptor( transaction, xxreg, candidateClients.getIds() ); // 
		
		if (iid>0){
			ix = candidateClients.indexOf(iid);
			candidateClients.remove(ix);
			// 
			out.print(2, "...receptor identified that also satisfies the context through its subscription, id:"+iid );
			instanceID = iid;
			// simply returning the iid will cause sending in 
			// SourceMessageInterpreter::sourceVariant_DataAcceptAndRelayToReceptor()
		}
		
		// for the rest, serve it as a subscription
		// preparing the subscription message, according to the settings
		// we will use a separate (but local) temporary thread for this job, such
		// that we can quickly return
		if (candidateClients.getIds().size()>0){
			
			SubscriptionSender subscriptSender = new SubscriptionSender( this, messageBoard, out);
		 
			subscriptSender.setData(  xtid, xiid, candidateClients, contextDescription ) ;
			subscriptSender.go();
		}
		
		// TODO send it also to other MessageBoards, yet not as a task delivery, but as a subscription! 
		candidateClients.clear();
		
		return instanceID ;
	}
	
	/**
	 * 
	 * 
	 * 
	 * @param futurePropertySet
	 * @param contextDescription
	 * @return
	 */
	@SuppressWarnings("unused")
	private boolean comparePlainContext( FutureProperties futurePropertySet, 
										 DistanceControlProperties dcp, 
										 ContextHeaderDescription contextDescription){    
	
		boolean rB=false, hb ;
		double miniMatchThreshold, miniSimThreshold;
		int n, cdataType=-1;
		
		String[] fcItemSet, fxItemSet ;
		double[] fcvalueset1;
		  
		ComparisonIntf comparison ;
		
		
		
		fcItemSet = futurePropertySet.values;
		fxItemSet = futurePropertySet.exclvalues;
		miniMatchThreshold = futurePropertySet.minimalMatch;
		
		miniSimThreshold = futurePropertySet.minimalProfileSimilarity ;
		
		fcvalueset1 = futurePropertySet.dataValues;
	
		n = fcItemSet.length ;
		
		try{
			
			// note that the CompareKernel can be set on the fly through the glue API

			//
			comparison = mbf.getComparisonFactory().createComparisonOrganizer( mbf.getCompareKernel(0) );
			cdataType = contextDescription.getCDataType() ;
			
			
			if ( cdataType == ContextInfra._DTYP_ITEMS){
				
				// now compare the item sets...
				rB = comparison.compareSets( dcp, fcItemSet, contextDescription.itemSet, fxItemSet, miniMatchThreshold) ;
	                 if (rB){
	                	 
	                	 compareResultHandle = (CompareResultHandle)comparison.getLastResultHandle();
	                	 
	                	 //CompareResultHandle compareResultHandle 
	                	 double   sdv = compareResultHandle.getSimilarityValue();
	                	 String[] commons = compareResultHandle.getSetItemsCommon();
	                	 String[] difference = compareResultHandle.getSetItemsDifference(1);
	                	 String   statusMsg = compareResultHandle.getLastStatusMessage();
	                 }
				/* optional, for further custom rules
				commons    = comparison.getCommonItems();
				difference = comparison.getDifferences(1);
				n = commons.length + difference.length ;
				*/
			}
			
			if ( cdataType == ContextInfra._DTYP_FVECTOR){
				rB = false;	 
				
				rB = comparison.compareFeatures( 	dcp, fcItemSet, fcvalueset1, 
											contextDescription.itemSet, contextDescription.getDataItems() , 
											fxItemSet,  // whether to exclude or not is in the interest of the receptor  
											miniMatchThreshold, miniSimThreshold ); // contextDescription.getMinimumSimilarity() ) ;
				                        //  perspective: RECEPTOR     
			}
			if ( cdataType == ContextInfra._DTYP_FULLTEXT){
				// here we have several possibilities, about which we should not decide here:
				// text-text, text-context, text-item set
				comparison.compareText( futurePropertySet.getFullText(), contextDescription  );
			}
			if ( cdataType == ContextInfra._DTYP_BINIMG){
				
			}
			
			
		}catch(Exception e){
			e.printStackTrace();
		}
		
		comparison=null;
		return rB;
	}

	private long findReceptor( Transaction transaction , XRegistration txreg, Vector<Long>  candidateClients ){
		
		long candidateID = -1, iid;
		int sv;
		boolean hb;
		String pmsRole;
		XRegistration xreg ;
		
		if (transaction.getRelayByContext() <= 0){
			return candidateID;
		}
		candidateID = -3;
		
		if (txreg==null){
			return -2;
		}
		
		pmsRole = getCoParticipant( txreg.getRole() );
		//??? none identified ? 
		
		if (candidateClients.size()>0){
			//
			for (int i=0;i<candidateClients.size();i++){
				
				iid = candidateClients.get(i) ;
				xreg = XRegistration.getItemByID( iid, registeredInstances) ;
				// for now, we simply take the first available one. no load balancing or other jokes here
				hb = xreg.getRole().toLowerCase().contentEquals(pmsRole.toLowerCase());
				if (hb){
					sv = xreg.getReceptorState();
					hb = (sv > XRegistration.RST_IS_STARTED) && (sv <= XRegistration.RST_IS_AVAIL);
						  
						 if (hb==false){ 
							 out.print(4, "");
							 if (sv == XRegistration.RST_IS_ALLOCATED){
								 
							 }
						}
				}
				if (hb){
					candidateID = iid;
					break;
				}
			}// i->
		}
		
		return candidateID;
	}
	
	/**
	 * 
	 * </br>
	 * if types of context and subscription do not match, it will return FALSE
	 *  
	 * @param contextDescription
	 * @param future
	 * @return
	 */
	private boolean matchingContextInfoToSubscription( ContextHeaderDescription contextDescription, Future future){
		
		boolean rB=false ;
		
		 
		
		//ContextIntf context;
		Vector<FutureProperties>  futureProps;
		FutureProperties futurePropertySet;
		
												out.print(3,"going to perform next comparison between contextDescription, future...");
		
		try{
			
			futureProps = future.getProperties() ;
			// context = (Context)future.getContext() ;
			
			
			for (int i=0;i<futureProps.size();i++){
				
				futurePropertySet = futureProps.get(i);
				
				if (futurePropertySet!=null){
					
					
					
					//context = futurePropertySet.context;  , 
					if (futurePropertySet.type != FutureProperties._CONTEXT){
						
						// in case of plain, simple context we just have to compare 2 sets (and the excluding items)
						// everything except full context is considered as "plain" here (only here!)
						rB = comparePlainContext( futurePropertySet, future.distanceCtrl, contextDescription);
					}
					 
					// contextDescription ->  _CSTYL_PLAIN    = 2; _CSTYL_COMPLETE = 3;
					if (futurePropertySet.type == FutureProperties._CONTEXT){
						// rB = compareFullContexts( futurePropertySet, contextDescription);
					}

				} // futurePropertySet != null ?
				
			} // i->
			
		
			
		}catch(Exception e){
			
		}
		
		
		return rB;
	}
	
	@SuppressWarnings("unused")
	private String getCoParticipant( String roleOfinitializer){
		String coparticipant = "";
		
		if (roleOfinitializer.toLowerCase().contains("source")) {
			coparticipant = "RECEPTOR";
		}
		if (roleOfinitializer.toLowerCase().contains("receptor")) {
			coparticipant = "SOURCE";
		}
		
		return coparticipant;
	}

	public  void relayToSubscriber( long[] iids , MMMessage mmMsg ){
		 
		// create a dedicated xml...
		// if the settings allow, also including the transaction again
		XRegistration xreg;
		
		long iid;
		
		for (int i=0;i<iids.length;i++){
			
			iid = iids[i];
			
			xreg = XRegistration.getItemByID( iid, registeredInstances) ;
											out.printErr(2, "relaying message to subscriber (iid:"+iid+")...") ;
			
		} // i->

		// and send it
		// SenderWrap sender = new SenderWrap (); 
		 
	}
	 
	
	/**
	 * this class takes a raw xml message and assimilates it as a "MMMessage" item 
	 * into the collection contextDescriptions;</br>
	 * the main thread loop of "MessageMatcherProcess{}" will work on this
	 * collection in FiFo-style </br>
	 * the class dealing with that is SubscriptionHandlingProcess{}
	 * 
	 * 
	 * @author kwa
	 *
	 */
	class AssimilateRawXml implements Runnable{  
		
		String rawXml, xmlMsg;
		String processedXmlStr="";
		
		boolean isRunning=true;
		
		MMMessage mmmsg;
		
		String mmmsgGuid = "" ;
		
		Thread arxThrd=null;
		
		
		public AssimilateRawXml( String rawxml, MMMessage mmsg){
			rawXml = rawxml;
			
			mmmsg = mmsg;
			arxThrd = new Thread(this, "arxThrd") ;
			arxThrd.start();
		}
		
		public String getMMMGuid(){
			return mmmsgGuid;
		}
		
		private void processXmlStr( String xmlstr){
			
			/*  context section in xml message
			     
			  	<context label="context A" style="1">
         			<pragma disclosures="data;name;" name="testSource0" role="SOURCE"/>
         			<items type="mandatory">
            			<list values="x;y;k"/>
            			<minimalmatch value="0.55"/>
         			</items>
      			</context>
			*/
			String str,contextXml="";
			// if there is no indication to keep the data section, we remove it...
			
			str = xmsg.getSpecifiedInfo(xmlstr, "//context/datadisclose", "value");
			
			if (rawXml.contains("<values>")){
				xmlMsg = xmsg.removeDataSection(rawXml, "//transaction/data/values");
			}else{
				xmlMsg = rawXml;
			}
			
			//xmlMsg = xmsg.removeTag( xmlMsg , "instance");
			xmlMsg = xmsg.removeTag( xmlMsg , "//transaction/name");
			xmlMsg = xmsg.removeTag( xmlMsg , "//transaction/request");
			 // does not work
		      
			if (xmlMsg.length()==0){
				return;
			}
			
			processedXmlStr = xmlMsg;
			
			mmmsg.xmlMsg = xmlMsg ;

			mmmsg.timeOfRecording = System.currentTimeMillis() ;
			
			mmmsg.ttl = 20000; 
			
			// create context object from context xml
			
			mmmsg.contextInfoFromMsg = cxmsg.readContextInfoFromXml( xmlMsg );
			
			if (mmmsg.contextInfoFromMsg != null){
				contextDescriptions.add( mmmsg );
			}
			mmmsgGuid = GUID.randomvalue() ;
		}
		
		public void go(){
			arxThrd.start();
		}
		public String get(){
			
			while (arxThrd==null){
				out.delay(1);
			}
			out.delay(5);
			while (isRunning){
				out.delay(1);
			}
			return processedXmlStr;
		}

		@Override
		public void run() {
			isRunning=true;
			
			processXmlStr( rawXml );
			
			isRunning=false;
		}
	}
	
}


class SubscriptionSender implements Runnable{
	
// object references ..............
	
	MessageBoard messageBoard ;
	MetaRegistration metaRegistration;
	MessageBoardFactory mbf;
	
	MsgBoardDataStorage msgBoardStorage;
	
	MessageMatcherProcess mmpParent;
	
	Vector<XRegistration> registeredInstances ;
	
	Vector<Long> candidateClients;
	Vector<Future> cFutures;
	
	ContextHeaderDescription contextDescription;
	long xtransactionID , xinstanceID;
	
	boolean isRunning;
	
	Thread subscriptSrvThrd;
	PrintLog out ;
	
	
	public SubscriptionSender( MessageMatcherProcess parent, MessageBoard msgBoard, PrintLog outprn){
	
		messageBoard = msgBoard;
		metaRegistration = messageBoard.getMetaRegistration() ;
		registeredInstances = metaRegistration.getRegisteredInstances() ; 
		
		mbf = messageBoard.getMsgBoardShares().getMsgBoardFactory() ;
		
		msgBoardStorage = messageBoard.getMsgBoardStorage() ;
		
		mmpParent = parent;
		
		subscriptSrvThrd = new Thread(this, "subscriptServeThrd"); 
		out = outprn;
	}
	
	public void setData( long xtid, long xiid, CClients candidates, ContextHeaderDescription cd){
		
		// need a copy !! since this thread will end soon...
		candidateClients = new Vector<Long>(candidates.getIds()) ;
		// also the attached info about futures...
		cFutures = new Vector<Future>(candidates.getFutures());
		
		xtransactionID = xtid; 
		xinstanceID = xiid ; 
		
		contextDescription = cd; // is not complete !!!
	}
	
	public void go(){
		subscriptSrvThrd.start();
	}
	
	private void send( String xmlstr, XRegistration xreg){
		int port=-1;
		
		// sending to client not to the board !!!
		SenderWrap sender = new SenderWrap( messageBoard ) ;
		sender.setWaitForPerformCompletion(false);
		sender.sendingMessage(xmlstr, xreg.getClientListeningPort());
		// sendingMessage(xmlstr, xreg);
		
	}
	
	private void perform(){
		long targetiid;
		String filename, rawXmlMsg, xmlMsg, insertionXml, namestr , nameDescend, futinfo = "";
		Future cfuture;
		XRegistration xreg;
		
		// these Id's (xtid, xiid) will identify the temporary file containing the whole transaction
		// we will read that, modify it and send it
		filename = msgBoardStorage.createTransactionDataObjTmpFilename( xtransactionID , xinstanceID );
		filename = filename.replace("/~tdata_", "/~srcreq_") ;
		rawXmlMsg = (String)msgBoardStorage.readTransactionDataObj( filename) ;
		
		nameDescend = contextDescription.getNameOfIssuer();
		
		// finally, relay-sending the transaction to all remaining subscribers, using the delivery pad
		for (int i=0;i<candidateClients.size();i++){
			
			targetiid = candidateClients.get(i) ;
			cfuture = cFutures.get(i);
			
			xreg = XRegistration.getItemByID(targetiid, registeredInstances);
												out.print(2, "...serving subscription to instance id:"+targetiid );
			xmlMsg = rawXmlMsg ;
			
 			if (contextDescription.disclosureAllowData==false){
// test manually if this is working....
 				xmlMsg = mmpParent.xmsg.removeDataSection(xmlMsg, "//transaction/data/values");
 			}
			
 			if ((contextDescription.disclosureAllowName==false) || (nameDescend.contentEquals("*"))){
 				namestr="";
 			}else{
 				// String str = xreg.getNameOfApplicant() ;
 				// remove uuid from that
 				
 				// create attribute value pair ready for insertion
 				namestr= "name=\""+nameDescend+"\"";
 			}
 			// removing the secret renders an answer impossible
 			// <request secret="634b45c0-28b3-11e1-88d2-e0f847024516" type="delivery"/>
 			// -> <request type="subscription"/>
 			xmlMsg = mmpParent.xmsg.changeAddSegment(xmlMsg, "//transaction/request", "request", "secret", "") ;
 			xmlMsg = mmpParent.xmsg.changeAddSegment(xmlMsg, "//transaction/request", "request", "type", "subscription") ;
 			
 			futinfo = "";
 			if (cfuture!=null){
 				// the subscriber maintains a map based on a guid, so this allows the subscriber to
 				// identify which future/subscription triggered the response
 				futinfo = cfuture.guidStr ;
 			}
 			insertionXml = "   <subscription source=\""+xinstanceID+"\" "+namestr+"  match=\""+futinfo+"\" />\n         ";
 			xmlMsg = mmpParent.xmsg.insertXmlStrToXmlStr(xmlMsg, "context", insertionXml); 
 			
 			send( xmlMsg, xreg) ;
		} // i->
	}

	@Override
	public void run() {
		 
		isRunning = true;
		
		perform() ;
		
		isRunning = false;
	}
}



class SubscriptionHandlingProcess implements Runnable{
	
	MessageMatcherProcess parent;
	String xmlMsg;
	
	MMMessage mmMsg ;
	
	Thread shpThrd;
	
	
	public SubscriptionHandlingProcess( MessageMatcherProcess parent, MMMessage mmmsg ){
		this.parent = parent ;
		parent.processCounter++ ;
		
		
		mmMsg = mmmsg;
		xmlMsg = mmMsg.xmlMsg ;
		
		shpThrd = new Thread(this, "shpThrd-"+parent.processCounter) ; 
		
		parent.__DEBUG_SHP_running=true;
 	}
	
	public void start(){
		shpThrd.start() ;
	}

	@Override
	public void run() {
		 
		boolean rB;
		long[] iids = new long[0];
		
		// given this particular message, find all subscribers and return their iid
		iids = parent.msgMatcher.checkforSubscriptions( mmMsg );
		
		if ((iids!=null) && (iids.length>0)){
			parent.relayToSubscriber( iids, mmMsg );
		}
		parent.processCounter-- ;
		
		parent.__DEBUG_SHP_running=false;
	}
}


class CClients{

	MessageBoardFactory mbf;
	Vector<Long> instanceIDs = new  Vector<Long>();
	Vector<Future> futures = new  Vector<Future>();
	
	public CClients( MessageBoardFactory mbf){
		this.mbf = mbf;
	}
	
	public void add( long iid){
		instanceIDs.add(iid);
		
		// also attached info, an empty position here
		
		Future f = new Future( mbf.getMsgBoardShare().getCompareKernel().getDistanceCtrl() );
		f.instanceID = -5;
		f.guidStr = "";
		futures.add(f) ;
	}
	
	public void add( long iid, Future future){

		instanceIDs.add(iid);
		// also attached info
		futures.add(future) ;
	}

	
	public Vector<Long> getIds(){
		return instanceIDs;
	}
	public Vector<Future> getFutures(){
		return futures;
	}
	
	
	public int indexOf( long iid){
		
		return instanceIDs.indexOf(iid);
	}
	
	
	public void remove(int index){
		
		instanceIDs.remove(index);
		futures.remove(index);
	}
	
	public void clear(){
		instanceIDs.clear();
		futures.clear();
		
		instanceIDs = null;
		futures = null;
	}
}

/**
 * 
 * MessageMatcherMessage: for internal decoupling of processes
 * 
 * 
 * @author kwa
 *
 */
class MMMessage{
	
	long iid=0, tid=0 ;
	
	String xmlMsg = "";
	long timeOfRecording=0, ttl=0;
	
	Context contextInfoFromMsg;
	
	
	public MMMessage(){
		
	}

	public long getIid() {
		return iid;
	}

	public void setIid(long iid) {
		this.iid = iid;
	}

	public long getTid() {
		return tid;
	}

	public void setTid(long tid) {
		this.tid = tid;
	}

	public String getXmlMsg() {
		return xmlMsg;
	}

	public void setXmlMsg(String xmlMsg) {
		this.xmlMsg = xmlMsg;
	}
}

